package mr;

import bean.DataPoint;
import bean.DelteData;
import com.htiiot.store.model.DeviceNumber;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.zookeeper.ZooKeeper;
import redis.clients.jedis.Jedis;
import tools.ZookeeperClient;
import util.SevenPointFunction;
import util.SevenPointFunctionV2;

import java.io.IOException;
import java.sql.*;
import java.text.DecimalFormat;
import java.util.*;

/**
 * @author :chensi
 */
public class DegradationAnalysis {
    private static final long ONE_MB = 1024 * 1024L;

    static class DegradationMapper extends Mapper<LongWritable, Text, Text, DataPoint> {
        private Jedis jedis = null;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            String address = context.getConfiguration().get("redis.address");
            String port = context.getConfiguration().get("redis.port");
            String auth = context.getConfiguration().get("redis.auth");
            jedis = new Jedis(address, Integer.parseInt(port));
            jedis.auth(auth);
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            jedis.close();
        }

        /**
         * 判断测点数据格式是否正确
         * @param data
         * @return
         */
        public boolean checkData(String[] data) {
            String tid = data[0].trim();
            String did = data[1].trim();
            String ts = data[2].trim();
            return StringUtils.isNotEmpty(tid) && StringUtils.isNotEmpty(did) && StringUtils.isNotEmpty(ts);
        }

        public boolean checkDataAndDn(String[] data, Set<String> dnSet) {
            return data.length == (dnSet.size() + 3);
        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] splits = line.split(",", -1);
            if (checkData(splits)) {
                String tid = splits[0].trim();
                String did = splits[1].trim();
                long ts = Long.parseLong(splits[2].trim());
                Set<String> dnSet = jedis.smembers("k2h:" + tid + ":" + did);
                if (checkDataAndDn(splits, dnSet)) {
                    List<String> dnList = new ArrayList<>(dnSet);
                    for (int i = 3; i < splits.length; i++) {
                        if (!"".equals(splits[i])) {
                            String dn = dnList.get(i - 3);
                            DataPoint dp = new DataPoint(dn, ts, Double.parseDouble(splits[i]));
                            context.write(new Text(dn), dp);
                        }
                    }
                } else {
                    System.err.println("Data and Dn not match!!!");
                    System.err.println("Data: " + Arrays.toString(splits));
                    System.err.println("DnSet: " + dnSet);
                }
            } else {
                System.err.println("Data format Exception!Data: " + Arrays.toString(splits));
            }
        }
    }

    static class DegradationReduce extends Reducer<Text, DataPoint, Text, LongWritable> {
        private Jedis jedis = null;
        private String redisBreakDownKey = "breakdowntime-";
        private Connection conn = null;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            redisBreakDownKey = redisBreakDownKey + context.getConfiguration().get("redisKeyTime");
            String address = context.getConfiguration().get("redis.address");
            String port = context.getConfiguration().get("redis.port");
            String auth = context.getConfiguration().get("redis.auth");
            jedis = new Jedis(address, Integer.parseInt(port));
            jedis.auth(auth);
            //初始化jdbc连接
            try {
                Class.forName(context.getConfiguration().get("jdbc.driverclassname"));
                conn = DriverManager.getConnection(context.getConfiguration().get("jdbc.url"), context.getConfiguration().get("jdbc.user"), context.getConfiguration().get("jdbc.password"));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (jedis != null) {
                jedis.close();
            }
        }

        @Override
        protected void reduce(Text key, Iterable<DataPoint> values, Context context) throws IOException, InterruptedException {
            List<DataPoint> data = new ArrayList<>();
            Iterator<DataPoint> it = values.iterator();
            while (it.hasNext()) {
                DataPoint dp = it.next();
                data.add(new DataPoint(dp.getDn(), dp.getTs(), dp.getValue()));
            }
            data.sort(new Comparator<DataPoint>() {
                @Override
                public int compare(DataPoint o1, DataPoint o2) {
                    return (int) (o1.getTs() - o2.getTs());
                }
            });
            //获取中值
            DeviceNumber dn = DeviceNumber.fromHexString(key.toString());
            long metricId = dn.getComponentIdLong();
            Statement stat = null;
            String midValue = null;
            try {
                stat = conn.createStatement();
                ResultSet rs = stat.executeQuery("select min,max from device_metric where id=" + metricId);
                while (rs.next()) {
                    Object min = rs.getObject("min");
                    Object max = rs.getObject("max");
                    if (min != null && max != null) {
                        midValue = new DecimalFormat("0.00").format((float) ((double) min + (double) max) / 2);
                    }
                }
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                if (stat != null) {
                    try {
                        stat.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }
            if (StringUtils.isNotEmpty(midValue)) {
                DataPoint[] sourceData = data.toArray(new DataPoint[data.size()]);
                System.out.println("SourceDataSize:" + sourceData.length);
                if (sourceData.length >= 7) {
                    System.out.println(Arrays.toString(sourceData));
                }
                //执行七点算法
                List<DelteData[]> l1 = new SevenPointFunction().sevenPointFunction(sourceData, Double.parseDouble(midValue), null);
                List<DataPoint[]> l2 = new SevenPointFunctionV2().process(sourceData, Double.parseDouble(midValue), null);
                //计算故障时间
                long allBreakDownTime = 0L;
                for (DelteData[] arr : l1) {
                    allBreakDownTime += arr[arr.length - 1].getSufMetricData().getTs() - arr[0].getPreMetricData().getTs();
                }
                for (DataPoint[] arr : l2) {
                    allBreakDownTime += arr[arr.length - 1].getTs() - arr[0].getTs();
                }
                if (allBreakDownTime != 0) {
                    jedis.hset(redisBreakDownKey, key.toString(), String.valueOf(allBreakDownTime));
                }
                context.write(key, new LongWritable(allBreakDownTime));
            } else {
                System.out.println("DN:" + key.toString() + " MetricId:" + metricId + " Do not have midValue!");
            }
        }
    }


    public static void main(String[] args) {
        if (args.length < 4 || "--help".equals(args[0]) || "-h".equals(args[0])) {
            System.out.println("Usage:\r\nhadoop jar <jarpath> <inputFilePath> <outputPath> <zookeeperAddress> <reduceNum>");
            System.exit(1);
        }
        String inputPath = args[0];
        String outputPath = args[1];
        String zkAddr = args[2];
        String reduceNum = args[3];
        String[] split = outputPath.split("/");
        String redisKeyTime = split[split.length - 1];
        Configuration conf = new Configuration();
        conf.set("mapreduce.input.fileinputformat.split.maxsize", String.valueOf(ONE_MB * 32));
        //从redis获取数据
        ZookeeperClient zkClient = new ZookeeperClient();
        ZooKeeper zk = zkClient.getZk(zkAddr, 30000);
        Properties redisProp = zkClient.getAll(zk, "/conf_htiiot/redis");
        Properties jdbcPro = zkClient.getAll(zk, "/conf_htiiot/maindb");
        //存入redis配置
        conf.set("redis.address", redisProp.getProperty("address"));
        conf.set("redis.port", redisProp.getProperty("port"));
        conf.set("redis.auth", redisProp.getProperty("auth"));
        conf.set("jdbc.url", jdbcPro.getProperty("url"));
        conf.set("jdbc.user", jdbcPro.getProperty("user"));
        conf.set("jdbc.password", jdbcPro.getProperty("sec.password"));
        conf.set("jdbc.driverclassname", jdbcPro.getProperty("driverclassname"));
        conf.set("redisKeyTime", redisKeyTime);
        FileSystem fs = null;
        try {
            Job job = Job.getInstance(conf);
            fs = FileSystem.get(conf);
            FileStatus[] fileStatuses = fs.listStatus(new Path(inputPath));
            Path[] paths = FileUtil.stat2Paths(fileStatuses);
            FileInputFormat.setInputPaths(job, paths);
            FileOutputFormat.setOutputPath(job, new Path(outputPath));
            job.setInputFormatClass(CombineTextInputFormat.class);
            job.setJarByClass(DegradationAnalysis.class);
            job.setMapperClass(DegradationMapper.class);
            job.setReducerClass(DegradationReduce.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(DataPoint.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(LongWritable.class);
            job.setNumReduceTasks(Integer.parseInt(reduceNum));
            job.waitForCompletion(true);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } finally {
            try {
                if (fs != null) {
                    fs.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
